package cn.me.alphamq.broker.event.spi.listener.broker;

import cn.me.alphamq.broker.cache.Container;
import cn.me.alphamq.broker.event.model.broker.ConsumeMsgEvent;
import cn.me.alphamq.broker.model.consumequeue.ConsumeQueueConsumeReq;
import cn.me.alphamq.broker.rebalance.ConsumerInstance;
import cn.me.alphamq.broker.rebalance.ConsumerInstanceManager;
import cn.me.alphamq.common.entity.req.ConsumeMsgReq;
import cn.me.alphamq.common.entity.resp.ConsumeMsgResp;
import cn.me.alphamq.common.enumeration.EventCodeEnum;
import cn.me.alphamq.common.event.Listener;
import cn.me.alphamq.common.msg.TcpMsg;
import com.alibaba.fastjson2.JSON;

import java.util.ArrayList;

/**
 * 消费消息事件的监听器
 *
 * @author f
 */
public class ConsumeMsgListener implements Listener<ConsumeMsgEvent> {

    @Override
    public void onReceive(ConsumeMsgEvent event) throws Exception {
        ConsumeMsgReq consumeMsgReq = event.getConsumeMsgReq();
        String ip = consumeMsgReq.getConsumerIp();
        Integer port = consumeMsgReq.getConsumerPort();
        String topic = consumeMsgReq.getTopic();
        String consumerGroup = consumeMsgReq.getConsumerGroup();
        Integer queueId = consumeMsgReq.getQueueId();
        Integer batchMsgNum = consumeMsgReq.getBatchMsgNum();

        // 构建消费者实例，从 重平衡策略 中分配消费者实例
        ConsumerInstance consumerInstance = ConsumerInstanceManager.getInstance(
                new ConsumerInstance(ip, port, topic, consumerGroup, queueId));

        ConsumeMsgResp consumeMsgResp = new ConsumeMsgResp();
        consumeMsgResp.setMsgId(consumeMsgReq.getMsgId());
        if (consumerInstance.getQueueIdList().stream().anyMatch(assignedQueueId -> assignedQueueId.equals(queueId))) {
            // 如果该消费者实例分配到它想要消费的队列，则进行消费
            consumeMsgResp.setMessageList(Container.getConsumeQueueConsumeHandler().consume(
                    new ConsumeQueueConsumeReq(topic, consumerGroup, queueId, batchMsgNum)
            ));
        } else {
            // 保证 bodyList 不为 null
            consumeMsgResp.setMessageList(new ArrayList<>());
        }

        event.getCtx().writeAndFlush(new TcpMsg(EventCodeEnum.CONSUME_MSG_RESP,
                JSON.toJSONBytes(consumeMsgResp)));
    }
}
